package com.hexiang.rocketmq.mq.producer;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

@Slf4j
@Component
public class ComplexBatchSpringMessageProducer {

    private static final String TOPIC = "test_complex_batch_spring_topic";

    @Resource
    private RocketMQTemplate rocketMQTemplate;

    @Value("${rocketmq.producer.max-message-size}")
    private Integer producerMaxMessageSize;

    @Resource
    private ObjectMapper objectMapper;

    public void sendComplexBatchMsg() {

        List<Message> messages = new ArrayList<>();
        for (int i = 0; i < 100; i++) {
            messages.add(new Message(TOPIC, "", "", ("Hello world " + i + "").getBytes()));
        }

        //把大的消息分裂成若干个小的消息
        ListSplitter splitter = new ListSplitter(messages);
        while (splitter.hasNext()) {
            try {
                List<Message> listItem = splitter.next();
                SendResult sendResult = rocketMQTemplate.syncSend(TOPIC, listItem);
                log.warn("sendResult " + sendResult);
            } catch (Exception e) {
                log.error("sendComplexBatchMsgErr", e);

            }
        }
    }

    class ListSplitter implements Iterator<List<Message>> {
        private final int SIZE_LIMIT = producerMaxMessageSize;
        private final List<Message> messages;
        private int currIndex;

        public ListSplitter(List<Message> messages) {
            this.messages = messages;
        }

        @Override
        public boolean hasNext() {
            return currIndex < messages.size();
        }

        @Override
        public List<Message> next() {
            int startIndex = getStartIndex();
            int nextIndex = startIndex;
            int totalSize = 0;
            for (; nextIndex < messages.size(); nextIndex++) {
                Message message = messages.get(nextIndex);
                int tmpSize = calcMessageSize(message);
                totalSize += tmpSize;
                if (totalSize > SIZE_LIMIT) {
                    break;
                }
            }
            List<Message> subList = messages.subList(startIndex, nextIndex);
            currIndex = nextIndex;
            return subList;
        }

        private int getStartIndex() {
            Message currMessage = messages.get(currIndex);
            int tmpSize = calcMessageSize(currMessage);
            while (tmpSize > SIZE_LIMIT) {
                currIndex += 1;
                Message message = messages.get(currIndex);
                tmpSize = calcMessageSize(message);
            }
            return currIndex;
        }

        private int calcMessageSize(Message message) {

            int tmpSize = 0;
            try {
                tmpSize = objectMapper.writeValueAsString(message).length();
            } catch (JsonProcessingException e) {
            }
            tmpSize = tmpSize + 20; // 增加⽇日志的开销20字节
            return tmpSize;
        }
    }

}